using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive.Concurrency;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Hosting;
using ReactiveChannelMessaging.Core;
using ReactiveChannelMessaging.Processors;

namespace ReactiveChannelMessaging.Services;

/// <summary>
/// 消息编排服务，协调各种消息流和处理器
/// </summary>
public class MessageOrchestrator : BackgroundService
{
    private readonly ReactiveMessageBus _messageBus;
    private readonly ILogger<MessageOrchestrator> _logger;
    private readonly List<IAsyncDisposable> _processors;
    private readonly Subject<string> _orchestrationEvents;

    // 消息处理器
    private DataMessageProcessor? _dataProcessor;
    private EventMessageProcessor? _eventProcessor;
    private CommandMessageProcessor? _commandProcessor;

    // 性能指标
    private readonly BehaviorSubject<PerformanceMetrics> _performanceMetrics;

    public MessageOrchestrator(
        ReactiveMessageBus messageBus,
        ILogger<MessageOrchestrator> logger)
    {
        _messageBus = messageBus;
        _logger = logger;
        _processors = new List<IAsyncDisposable>();
        _orchestrationEvents = new Subject<string>();
        _performanceMetrics = new BehaviorSubject<PerformanceMetrics>(new PerformanceMetrics());
    }

    /// <summary>
    /// 编排事件流
    /// </summary>
    public IObservable<string> OrchestrationEvents => _orchestrationEvents.AsObservable();

    /// <summary>
    /// 性能指标流
    /// </summary>
    public IObservable<PerformanceMetrics> PerformanceMetrics => _performanceMetrics.AsObservable();

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            await InitializeProcessorsAsync();
            await SetupMessageFlowsAsync();
            await SetupMonitoringAsync();

            _orchestrationEvents.OnNext("MessageOrchestrator started successfully");
            _logger.LogInformation("Message orchestration started");

            // 保持服务运行
            await Task.Delay(Timeout.Infinite, stoppingToken);
        }
        catch (OperationCanceledException)
        {
            _logger.LogInformation("Message orchestration stopped");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Fatal error in message orchestration");
            throw;
        }
    }

    private async Task InitializeProcessorsAsync()
    {
        // 初始化数据处理器
        _dataProcessor = new DataMessageProcessor(
            LoggerFactory.Create(builder => builder.AddConsole())
                .CreateLogger<DataMessageProcessor>());
        _processors.Add(_dataProcessor);

        // 初始化事件处理器
        _eventProcessor = new EventMessageProcessor(
            LoggerFactory.Create(builder => builder.AddConsole())
                .CreateLogger<EventMessageProcessor>());
        _processors.Add(_eventProcessor);

        // 初始化命令处理器
        _commandProcessor = new CommandMessageProcessor(
            LoggerFactory.Create(builder => builder.AddConsole())
                .CreateLogger<CommandMessageProcessor>(),
            _messageBus);
        _processors.Add(_commandProcessor);

        _logger.LogDebug("All message processors initialized");
        await Task.CompletedTask;
    }

    private async Task SetupMessageFlowsAsync()
    {
        // 设置数据消息流 - 支持背压控制
        _dataProcessor!.Subscribe(
            _messageBus.DataMessages
                .ObserveOn(Scheduler.Default)
                .Where(msg => !string.IsNullOrEmpty(msg.Content)),
            concurrency: 5);

        // 设置事件消息流 - 按事件类型分组处理
        _eventProcessor!.Subscribe(
            _messageBus.EventMessages
                .GroupBy(msg => msg.EventType)
                .SelectMany(group => group.ObserveOn(Scheduler.Default)),
            concurrency: 3);

        // 设置命令消息流 - 串行处理确保顺序
        _commandProcessor!.Subscribe(
            _messageBus.CommandMessages
                .ObserveOn(Scheduler.Default),
            concurrency: 1);

        // 设置复杂的消息流编排
        await SetupAdvancedFlowsAsync();

        _logger.LogDebug("Message flows configured");
    }

    private async Task SetupAdvancedFlowsAsync()
    {
        // 1. 数据聚合流 - 将相同源的数据消息聚合
        _messageBus.DataMessages
            .GroupBy(msg => msg.Source)
            .SelectMany(group => 
                group.Buffer(TimeSpan.FromSeconds(5), 10) // 5秒或10条消息触发聚合
                    .Where(buffer => buffer.Count > 0)
                    .Select(buffer => new 
                    {
                        Source = group.Key,
                        Count = buffer.Count,
                        TotalPriority = buffer.Sum(m => m.Priority),
                        Messages = buffer
                    }))
            .Subscribe(
                aggregation =>
                {
                    _logger.LogInformation(
                        "Data aggregated from {Source}: {Count} messages, total priority: {Priority}",
                        aggregation.Source, aggregation.Count, aggregation.TotalPriority);
                    
                    _orchestrationEvents.OnNext(
                        $"Aggregated {aggregation.Count} messages from {aggregation.Source}");
                },
                error => _logger.LogError(error, "Error in data aggregation flow"));

        // 2. 错误恢复流 - 自动重试失败的操作
        _messageBus.Errors
            .Where(ex => ex is not OperationCanceledException)
            .Sample(TimeSpan.FromSeconds(10)) // 防止错误风暴
            .Subscribe(
                error =>
                {
                    _logger.LogWarning("Implementing error recovery for: {Error}", error.Message);
                    _orchestrationEvents.OnNext($"Error recovery triggered: {error.GetType().Name}");
                },
                error => _logger.LogError(error, "Error in error recovery flow"));

        // 3. 响应关联流 - 追踪请求-响应对
        var pendingCommands = new Dictionary<Guid, DateTime>();
        
        _messageBus.CommandMessages
            .Subscribe(cmd =>
            {
                pendingCommands[cmd.Id] = DateTime.UtcNow;
                _logger.LogDebug("Command registered: {CommandId}", cmd.Id);
            });

        _messageBus.ResponseMessages
            .Subscribe(response =>
            {
                if (pendingCommands.TryGetValue(response.CorrelationId, out var startTime))
                {
                    var duration = DateTime.UtcNow - startTime;
                    pendingCommands.Remove(response.CorrelationId);
                    
                    _logger.LogInformation(
                        "Command-Response completed: {CommandId}, Duration: {Duration}ms, Success: {Success}",
                        response.CorrelationId, duration.TotalMilliseconds, response.Success);
                    
                    _orchestrationEvents.OnNext(
                        $"Command completed in {duration.TotalMilliseconds:F1}ms");
                }
            });

        // 4. 性能监控流
        Observable.Interval(TimeSpan.FromSeconds(30))
            .SelectMany(_ => _messageBus.GetMessageStats(TimeSpan.FromMinutes(1)))
            .Subscribe(stats =>
            {
                var totalMessages = stats.MessageCounts.Values.Sum();
                _performanceMetrics.OnNext(new PerformanceMetrics
                {
                    Timestamp = stats.Timestamp,
                    TotalMessages = totalMessages,
                    MessageBreakdown = stats.MessageCounts,
                    PendingCommands = pendingCommands.Count
                });
                
                _logger.LogInformation("Performance metrics updated: {TotalMessages} messages processed", 
                    totalMessages);
            });

        await Task.CompletedTask;
    }

    private async Task SetupMonitoringAsync()
    {
        // 健康状态监控
        _messageBus.HealthStatus
            .DistinctUntilChanged()
            .Subscribe(isHealthy =>
            {
                var status = isHealthy ? "Healthy" : "Unhealthy";
                _logger.LogInformation("Message bus health status: {Status}", status);
                _orchestrationEvents.OnNext($"Health status changed to: {status}");
            });

        // 事件类型趋势分析
        _eventProcessor!.EventTypes
            .Buffer(TimeSpan.FromMinutes(1))
            .Where(eventTypes => eventTypes.Count > 0)
            .Subscribe(eventTypes =>
            {
                var trends = eventTypes
                    .GroupBy(et => et)
                    .ToDictionary(g => g.Key, g => g.Count());
                
                _logger.LogInformation("Event trends (1min): {Trends}", 
                    string.Join(", ", trends.Select(kvp => $"{kvp.Key}: {kvp.Value}")));
            });

        await Task.CompletedTask;
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        _orchestrationEvents.OnNext("MessageOrchestrator stopping");
        
        foreach (var processor in _processors)
        {
            await processor.DisposeAsync();
        }
        
        _orchestrationEvents.Dispose();
        _performanceMetrics.Dispose();
        
        await base.StopAsync(cancellationToken);
    }
}

/// <summary>
/// 性能指标
/// </summary>
public class PerformanceMetrics
{
    public DateTime Timestamp { get; set; } = DateTime.UtcNow;
    public int TotalMessages { get; set; }
    public Dictionary<string, int> MessageBreakdown { get; set; } = new();
    public int PendingCommands { get; set; }
    public double MessagesPerSecond => TotalMessages / 60.0; // 基于1分钟窗口
}